package com.bd03.cmccmonitor.dbSql

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import scalikejdbc._

object SQLHandler {
  /**
   * 当天的统计结果入库
   * @param generalDayRes
   */
  def generalDay2Mysql(generalDayRes:RDD[(String, (Int, Int, Long, Int))])={
    generalDayRes.foreachPartition(it=>{
      DB.localTx(implicit session=>{
        it.foreach(t=>{
          sql"""
               |insert into cmcc_general(day,total,succ,costTime,money)
               |        values(?,?,?,?,?) on duplicate key
               |        update total=total+?,succ=succ+?,costTime=costTime+?,money=money+?
               |""".stripMargin.bind(t._1,t._2._1,t._2._2,t._2._3,t._2._4,t._2._1,t._2._2,t._2._3,t._2._4)
            .update().apply()
        })
      })
    })
  }

  /**
   * 每小时的的统计结果入库
   * @param generalHourRes
   */
  def generalHour2Mysql(generalHourRes: RDD[((String, String), (Int, Int, Long, Int))])={
    generalHourRes.foreachPartition(it=>{
      DB.localTx(implicit session=>{
        it.foreach(t=>{
          sql"""
               |insert into cmcc_general_hour(day,hour,total,succ)
               |        values(?,?,?,?) on duplicate key
               |        update total=total+?,succ=succ+?
               |""".stripMargin.bind(t._1._1,t._1._2,t._2._1,t._2._2,t._2._1,t._2._2)
            .update().apply()
        })
      })
    })
  }


  /**
   * 将当天各省份的总订单和成功充值订单入库
   * @param dayProvinceSuccRes
   */
  def provicneDay2Mysql(PCODE2PNAME: Broadcast[Map[String, AnyRef]], dayProvinceSuccRes: RDD[((String, String), (Int, Int))])={
    dayProvinceSuccRes.foreachPartition(it=>{
      DB.localTx(implicit session=>{
        it.foreach(t=>{
          sql"""
               |insert into cmcc_province_day(day,pname,pcode,total,succ)
               |        values(?,?,?,?,?) on duplicate key
               |        update total=total+?,succ=succ+?
               |""".stripMargin.bind(t._1._1,PCODE2PNAME.value.getOrElse(t._1._2,"未知"),t._1._2,t._2._1,t._2._2,t._2._1,t._2._2)
            .update().apply()
        })
      })
    })
  }

  /**
   * 每分钟充值笔数及金额入库
   * @param perMinutesResult
   */
  def perMinutes2MySQL(perMinutesResult: RDD[((String, String, String), (Int, Int))]) = {
    perMinutesResult.foreachPartition(iter => {
      DB.localTx { implicit session =>
        iter.foreach(tp => {
          SQL(
            """
              |INSERT INTO cmcc_perminutes (`day`, hour, minutes, succ, money)
              |VALUES (?, ?, ?, ?, ?) ON DUPLICATE KEY
              |UPDATE succ = succ + ?, money = money + ?
            """.stripMargin)
            .bind(
              tp._1._1,
              tp._1._2,
              tp._1._3,
              tp._2._1,
              tp._2._2,
              tp._2._1,
              tp._2._2
            ).update().apply()
        })
      }
    })
  }
}
